package jv.redis.datatype.queue;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import redis.clients.jedis.Jedis;
/*
 * 优先级队列
 */
public class RedisPriorityQueue {
	Jedis jedis;
	ExecutorService es;
	public RedisPriorityQueue(){
		initRedis();
		es = Executors.newCachedThreadPool();
	}
	

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		RedisPriorityQueue rqn = new RedisPriorityQueue();
		List<Future> lf = new ArrayList<Future>();
		lf.add(rqn.producerOne());
		lf.add(rqn.producerTwo());
		lf.add(rqn.consumer());
		for(Future f:lf){
			f.get();
		}
		rqn.close();
		System.out.println("program is finished");
	}
	
	public Future producerOne(){
		System.out.println("submit producerOne");
		return es.submit(new Runnable() {
				@Override
				public void run() {
					Jedis myEdis = newRedis();
					myEdis.del("redis:queue:one");
					for(int i=1;i<=1000;i++){
						myEdis.lpush("redis:queue:one", "test"+i);
						//System.out.println("producer:test"+i);
					}
					System.out.println("producer is finished");
				}
			});
	}
	public Future producerTwo(){
		System.out.println("submit producerTwo");
		return es.submit(new Runnable() {
				@Override
				public void run() {
					Jedis myEdis = newRedis();
					Random rd = new Random();
					int timeout = 0;
					myEdis.del("redis:queue:two");
					for(int i=1;i<=10;i++){
						myEdis.lpush("redis:queue:two", "test"+i);
						timeout = rd.nextInt(10);
						System.out.println("producer:test"+i+",sleep "+timeout+" seconds");
						try {
							Thread.sleep(timeout*1000);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
					System.out.println("producer is finished");
				}
			});
	}
	
	public Future consumer() {
		System.out.println("submit consumer");
		return es.submit(new Runnable() {
			@Override
			public void run() {
				Jedis myEdis = newRedis();
				List<String> rlt = null;
				/**
				 * 为了看得效果更明显，虽然采用了阻塞方式获取数据，还是强制每隔一秒休眠一次，以达到队列一种有很多消息，队列二中随机来消息，但是先取走
				 */
				while (true) {
					rlt = myEdis.brpop(0,"redis:queue:two","redis:queue:one");
					System.out.println("consumer:" + rlt);
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		});
	}
	
	public void initRedis(){
		jedis = new Jedis("redis://192.168.0.114:6379");
		System.out.println("redis is connected");
	}
	
	public Jedis newRedis(){
		return new Jedis("redis://192.168.0.114:6379");
	}

	public void closeRedis(){
		//注意不要使用jedis.shutdown()，这样会把服务器停掉。。。
		jedis.close();
	}
	
	public void closePool(){
		es.shutdown();
	}
	
	public void close(){
		closeRedis();
		closePool();
	}
}
